package tqf.test.first.treadcore;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author: Qifeng Tang
 * @date: 2022/9/1 10:03
 * @Version 1.0
 **/
public class Executor {

    //每个线程每次查询的条数
    private static final Integer LIMIT = 20;
    //起的线程数
    private static final Integer THREAD_NUM = 10;
    //创建线程池
    ThreadPoolExecutor pool = new ThreadPoolExecutor(THREAD_NUM, THREAD_NUM * 2, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100));

    public void push() {

        Integer dataCount=12000;
        //计数器，保证线程安全
        int threadCount = 0;
        //统计推送成功的消息数
        int totalSuccessCount = 0;

        int num = dataCount / (LIMIT * THREAD_NUM)+1 ;
        System.out.println("总共需要推送："+num+" 轮次");
        for(int i=0;i<num; i++) {
            //接收线程返回结果
            List<Future<Integer>> futureList = new ArrayList<>(32);
            //起threadNum个线程并行查询库，加锁
            for (int j = 0; j < THREAD_NUM; j++) {
                synchronized (this) {
                    threadCount++;
                    //提交线程，用数据起始位置标识线程
                    Future<Integer> future = pool.submit(new PushBrandUserTask(threadCount,LIMIT));
                    //先不取值，防止阻塞,放进集合
                    futureList.add(future);
                    System.out.println("页面起始位置："+threadCount);
                }
            }
            //统计本轮推送成功数据
            for (Future f : futureList) {
                try {
                    if(f.get() != null){
                        totalSuccessCount = totalSuccessCount + (int) f.get();
                    }
                    System.out.println("推送成功了："+totalSuccessCount);
                } catch (Exception e){
                    System.out.println("异常了："+e);
                }
            }
        }


    }
}
